package consumer

import (
	"errors"
	"fmt"
	"github.com/alecthomas/log4go"
	"github.com/streadway/amqp"
	"time"
)

type MQConsumer interface {
	Start() error
	Stop() error
}

func New(config Config, receiveChan chan<- []byte) MQConsumer {
	return &consumer{
		Config:          config,
		receiveChan:     receiveChan,
		notifyClose:     make(chan *amqp.Error),
		closeChan:       make(chan bool),
		stopConsumeChan: make(chan bool),
	}
}

type Config struct {
	Uri          string
	ExchangeName string
	ExchangeType string
	QueueName    string
	BindingKey   string
	ConsumerTag  string
}

type consumer struct {
	Config
	receiveChan                chan<- []byte
	conn                       *amqp.Connection
	channel                    *amqp.Channel
	deliveries                 <-chan amqp.Delivery
	notifyClose                chan *amqp.Error
	closeChan, stopConsumeChan chan bool
	connOk                     bool //连接状态
}

//
//func (this *consumer) Name() string {
//	return "rabbitmq consumer"
//}
//func (this *consumer) Init() {
//	var err error
//	if this.uri, err = config.GetStringDefault("rabbitmq", "uri", "amqp://guest:guest@localhost:5672/"); err != nil {
//		panic(fmt.Sprintf("%s，获取rabbitmq配置'uri'失败:%s", this.Name(),err.Error()))
//	}
//	if this.exchangeName,err=config.GetStringDefault("rabbitmq", "exchangeName", "test-exchange");err!=nil{
//		panic(fmt.Sprintf("%s，获取rabbitmq配置'exchangeName'失败:%s",this.Name(), err.Error()))
//	}
//	if this.exchangeType,err=config.GetStringDefault("rabbitmq","exchangeType","direct");err!=nil{
//		panic(fmt.Sprintf("%，获取rabbitmq配置'exchangeType'失败:%s",this.Name(), err.Error()))
//	}
//	if this.queueName,err=config.GetStringDefault("rabbitmq","queueName","test-queue");err!=nil{
//		panic(fmt.Sprintf("%，获取rabbitmq配置'queueName'失败:%s",this.Name(), err.Error()))
//	}
//	if this.bindingKey,err=config.GetStringDefault("rabbitmq","bindingKey","test-key");err!=nil{
//		panic(fmt.Sprintf("%s，获取rabbitmq配置'routingKey'失败:%s",this.Name(), err.Error()))
//	}
//	if this.consumerTag,err=config.GetStringDefault("rabbitmq","consumerTag","simple-consumer");err!=nil{
//		panic(fmt.Sprintf("%s，获取rabbitmq配置'consumerTag'失败:%s",this.Name(), err.Error()))
//	}
//	this.notifyClose = make(chan *amqp.Error)
//	this.closeChan = make(chan bool)
//	this.stopConsumeChan = make(chan bool)
//
//}
func (this *consumer) Start() error {
	go func() {
		this.begin()
	}()
	return nil
}
func (this *consumer) Stop() error {
	if !this.connOk {
		return errors.New(fmt.Sprintf("rabbitmq consumer,已关闭"))
	}
	//关闭连接的同时，channel也会关闭
	err := this.conn.Close()
	if err != nil {
		return err
	}
	this.connOk = false
	this.closeChan <- true
	close(this.closeChan)
	return nil
}
func (this *consumer) connect() (err error) {
	if this.connOk {
		return errors.New(fmt.Sprintf("rabbitmq consumer,已有连接并且连接正常，无需重复创建"))
	}
	this.conn, err = amqp.Dial(this.Uri)
	if err != nil {
		log4go.Error("rabbitmq consumer,创建连接失败: %s", err)
		return err
	} else {
		log4go.Info("rabbitmq consumer,创建rabbitmq消费者连接成功")
	}
	//创建channel
	this.channel, err = this.conn.Channel()
	if err != nil {
		log4go.Error("rabbitmq consumer,获取channel失败: %s", err)
		return err
	}
	//创建交换机
	if err = this.exchangeDeclare(); err != nil {
		log4go.Error("rabbitmq consumer,创建exchange失败: %s", err)
		return err
	}
	//创建队列
	if err = this.queueDeclare(); err != nil {
		log4go.Error("rabbitmq consumer,创建queue失败: %s", err)
		return err
	}
	//绑定队列
	if err = this.bindQueue(); err != nil {
		log4go.Error("rabbitmq consumer,绑定queue失败: %s", err)
		return err
	}

	this.connOk = true
	return nil
}
func (this *consumer) begin() {
	log4go.Info("rabbitmq consumer,创建连接并开启异常监听")
	for {
		this.connOk = false
		for {
			if err := this.connect(); err != nil {
				log4go.Error("rabbitmq consumer,创建连接失败：%s,准备重试", err.Error())
				time.Sleep(10 * time.Second)
				continue
			} else {
				//notifyClose已关闭，需重新定义和绑定
				this.notifyClose = make(chan *amqp.Error)
				this.channel.NotifyClose(this.notifyClose)
				//创建接收
				go func() {
					if err = this.pull(); err != nil {
						log4go.Error("rabbitmq consumer开启接收失败：%s", err)
						return
					}
				}()
				break
			}
		}
		select {
		case <-this.closeChan:
			this.stopConsumeChan <- true
			return
		case <-this.notifyClose:
			this.stopConsumeChan <- true
			break
		}
	}
}
func (this *consumer) exchangeDeclare() (err error) {
	if err = this.channel.ExchangeDeclare(
		this.ExchangeName, // name
		this.ExchangeType, // type
		true,              // durable
		false,             // auto-deleted
		false,             // internal
		false,             // noWait
		nil,               // arguments
	); err != nil {
		log4go.Error("rabbitmq consumer,创建exchange失败: %s", err.Error())
		return err
	}
	return nil
}
func (this *consumer) queueDeclare() (err error) {
	if _, err = this.channel.QueueDeclare(
		this.QueueName, // name of the queue
		true,           // durable
		false,          // delete when unused
		false,          // exclusive
		false,          // noWait
		nil,            // arguments
	); err != nil {
		log4go.Error("rabbitmq consumer,创建queue失败: %s", err.Error())
		return
	}
	return nil
}
func (this *consumer) bindQueue() (err error) {
	if err = this.channel.QueueBind(
		this.QueueName,    // name of the queue
		this.BindingKey,   // bindingKey
		this.ExchangeName, // sourceExchange
		false,             // noWait
		nil,               // arguments
	); err != nil {
		log4go.Error("rabbitmq consumer,绑定队列异常: %s", err.Error())
		return err
	}
	return nil
}

func (this *consumer) pull() (err error) {
	log4go.Info("rabbitmq consumer,创建接受者")
	this.deliveries, err = this.channel.Consume(
		this.QueueName,   // name
		this.ConsumerTag, // consumerTag,
		true,             // noAck
		false,            // exclusive
		false,            // noLocal
		false,            // noWait
		nil,              // arguments
	)
	if err != nil {
		log4go.Error("rabbitmq consumer,创建接收者异常：%s", err.Error())
		return err
	}
	for {
		select {
		case <-this.stopConsumeChan:
			log4go.Error("rabbitmq consumer,退出消费")
			return nil //this.channel.Cancel(this.consumerTag,false)
		case d := <-this.deliveries:
			//log4go.Info("%s,接收到：%s", this.Name(), string(d.Body))
			this.receiveChan <- d.Body
			//	d.Ack(true)//接收到就删除
		}
		//time.Sleep(1*time.Second)
	}
	return nil
}
